package club.babyf.reactor;

import reactor.core.publisher.Flux;

/**
 * Test.
 *
 * @author zhanglei.
 * @date 2025/1/15 11:29.
 * @description 测试类.
 */
public class Test {

    public static void main(String[] args) {
        // Mono: 表示包含 0 或 1 个元素的异步数据流，常用于处理单值（如 HTTP 请求的响应、数据库查询返回的单条数据）
//        Mono<String> mono = Mono.just("Hello, Reactor!");
//        // 输出: Hello, Reactor!
//        mono.subscribe(System.out::println);

        // Flux: 表示包含 0 到 N 个元素的异步数据流，常用于处理多值（如事件流、数据库查询返回的多条数据）
//        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
//        // 输出: 1 2 3 4 5
//        flux.subscribe(System.out::println);

        // 转换数据
//        // 输出: 2 4 6
//        Flux.just(1, 2, 3)
//                .map(i -> i * 2)
//                .subscribe(System.out::println);

        // 异步转换数据
//        // 输出: A1 B1
//        Flux.just("A", "B")
//                .flatMap(s -> Mono.just(s + "1"))
//                .subscribe(System.out::println);

        // 过滤数据
//        // 输出: 2 4 6 8 10
//        Flux.range(1, 10)
//                .filter(i -> i % 2 == 0)
//                .subscribe(System.out::println);

        // 合并多个数据流
//        // 输出: 1 2 3 4
//        Flux.merge(Flux.just(1, 2), Flux.just(3, 4))
//                .subscribe(System.out::println);

        // 合并两个数据流，并按索引位置组合
//        // 输出: [1, A] [2, B]
//        Flux.zip(Flux.just(1, 2), Flux.just("A", "B"))
//                .subscribe(System.out::println);

        // Schedulers（调度器）
        /*
              Reactor 提供了多种调度器，用于控制数据流的执行线程。
                Schedulers.boundedElastic: 适用于 I/O 阻塞操作。
                Schedulers.parallel: 适用于并行计算。
                Schedulers.single: 使用单一线程。
                Schedulers.immediate: 当前线程执行。
                Schedulers.fromExecutor: 使用自定义线程池。
         */
//        Flux.range(1, 10)
//                .publishOn(Schedulers.parallel())
//                .map(i -> i * 2)
//                .subscribe(System.out::println);

        // Backpressure（背压）
        /*
              Reactor 基于 Reactive Streams，支持背压机制：
                背压允许消费者控制数据流的速率，防止因生产者过快发射数据而导致内存溢出。
                Reactor 的背压策略包括：
                    Request n: 消费者请求 n 个数据。
                    Cancel: 取消订阅。
         */
        Flux.range(1, 1000)
                .onBackpressureBuffer(100)
                .subscribe(System.out::println);
    }

}
